package com.taobao.analyse.reduce;

import java.io.IOException;
import java.util.Iterator;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.analyse.constants.SplitConstants;

public class CategoryPriceChangeRateReducer extends MapReduceBase implements
		Reducer<LongWritable, Text, LongWritable, Text> {

	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {
		String category = "";
		String timeList = "";
		String categoryName = "";
		StringBuilder sb = new StringBuilder();
		while (values.hasNext()) {
			String info = values.next().toString();
			if (info.startsWith("count")) {
				sb.append(info.split(SplitConstants.SPLIT_AT)[1]);
			} else if (info.startsWith("category")) {
				category = getCategory(info);
			} else if (info.startsWith("detail")) {
				// 商品价格变动幅度怎么算？
			} else if (info.startsWith("time")) {
				//时间处理，可以单独写一个reduer，输出到其它文件中
				timeList = info;
			} else if(info.startsWith("categoryName")){
				categoryName = info;
			}
		}
//		if(StringUtils.isEmpty(category)){
//			return;
//		}
		
		try{
			long categoryId=Long.valueOf(category);
			String result = sb.toString();
			if(!StringUtils.isEmpty(result)){
				output.collect(new LongWritable(categoryId), new Text(result));
			}
			if(!StringUtils.isEmpty(timeList)){
				output.collect(new LongWritable(categoryId), new Text(timeList));
			}
			if(!StringUtils.isEmpty(categoryName)){
				output.collect(new LongWritable(categoryId), new Text(categoryName));
			}
		}catch(Exception e){
			//do nothing
		}
		
		//这一次还不应该计算
		/**
		Set<String> dates = updateCount.keySet();
		StringBuilder sb_dateList = new StringBuilder();
		StringBuilder sb_changeRate = new StringBuilder();
		DecimalFormat  df =new DecimalFormat("#.00000");  
		double priceChangeCount = 0;
		double totalEditCount = 0;
		for(String date : dates){
			sb_dateList.append(date).append(SplitConstants.SPLIT_OUTPUT);
			priceChangeCount = updatePriceCount.get(date);
			totalEditCount = updateCount.get(date);
			sb_changeRate.append(df.format(priceChangeCount/totalEditCount)).append(SplitConstants.SPLIT_OUTPUT);
		}
		try{
			long categoryId=Long.valueOf(category);
			output.collect(new LongWritable(categoryId), new Text(sb_dateList.toString()));
			output.collect(new LongWritable(categoryId), new Text(sb_changeRate.toString()));
		}catch(Exception e){
			//do nothing
		}
		**/
	}

	/**
	 * 获取类目id信息(现在只返回一级类目)
	 * 
	 * @param string
	 * @return
	 */
	private String getCategory(String data) {
		String[] catList = data.split(SplitConstants.SPLIT_AT);
		
		String[] cats = catList[1].split(SplitConstants.SPLIT_UNDER_BAR);
		if(cats.length < 1){
			return null;
		}
		return cats[0];
	}

}
